#pragma once
#include <vector>
#include <string>
#include <algorithm>
#include <functional>
#include <unordered_map>
#include <thread>
#include <memory>
#include <mutex>
#include <typeinfo>
#include <condition_variable>

#include <cstdint>
#include <cassert>
#include <ctime>
#include <cstdio>
#include <cstring>
#include <cerrno>
#include <cstdlib>

#include <unistd.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <fcntl.h>
#include <sys/epoll.h>
#include <sys/eventfd.h>
#include <sys/timerfd.h>
#include <signal.h>

#define INFO 1
#define DEBUG 2
#define ERROR 3
#define LOG_LEVEL INFO

#define LOG(level, format, ...)                                                                                       \
    do                                                                                                                \
    {                                                                                                                 \
        if (level < LOG_LEVEL)                                                                                        \
            break;                                                                                                    \
        time_t now = time(nullptr);                                                                                   \
        tm *t = localtime(&now);                                                                                      \
        char time_str[64] = {0};                                                                                      \
        strftime(time_str, sizeof(time_str), "%Y-%m-%d %H:%M:%S", t);                                                 \
        fprintf(stdout, "[%lu][%s][%s, %d]" format "\n", pthread_self(), time_str, __FILE__, __LINE__, ##__VA_ARGS__); \
    } while (false)

#define INFO_LOG(format, ...) LOG(INFO, format, ##__VA_ARGS__)
#define DEBUG_LOG(format, ...) LOG(DEBUG, format, ##__VA_ARGS__)
#define ERROR_LOG(format, ...) LOG(ERROR, format, ##__VA_ARGS__)

class Buffer
{
    const static uint64_t BUFFER_DEFAULT_SIZE = 1024;

public:
    Buffer(uint64_t size = BUFFER_DEFAULT_SIZE) : _buffer(size), _read_pos(0), _write_pos(0) {}
    Buffer(const void *data, int size)
        : _buffer(BUFFER_DEFAULT_SIZE), _read_pos(0), _write_pos(0)
    {
        EnsureWriteSize(size);
        WriteAndPush(data, size);
    }
    // 获取缓冲区首地址
    char *Begin() { return _buffer.data(); }
    const char *Begin() const { return _buffer.data(); }
    // 获取有效数据尾地址
    char *EffectiveEnd() { return _buffer.data() + _write_pos; }
    const char *EffectiveEnd() const { return _buffer.data() + _write_pos; }
    // 获取写指针
    char *WritePos() { return _buffer.data() + _write_pos; }
    const char *WritePos() const { return _buffer.data() + _write_pos; }
    // 获取读指针
    char *ReadPos() { return _buffer.data() + _read_pos; }
    const char *ReadPos() const { return _buffer.data() + _read_pos; }
    // 尾部空闲空间大小
    uint64_t TailIdleSize() { return _buffer.size() - _write_pos; }
    // 头部空闲空间大小
    uint64_t HeadIdleSize() { return _read_pos; }
    // 空闲空间大小
    uint64_t IdleSize() { return TailIdleSize() + HeadIdleSize(); }
    // 有效数据大小
    uint64_t EffectiveDataSize() const { return _write_pos - _read_pos; }
    // 移动读指针
    void MoveReadOffset(uint64_t offset)
    {
        assert(offset <= EffectiveDataSize());
        _read_pos += offset;
    }

    // 移动写指针
    void MoveWriteOffset(uint64_t offset)
    {
        assert(offset <= TailIdleSize());
        _write_pos += offset;
    }

    // 确保写入空间足够, 传入size为需要写入的大小
    void EnsureWriteSize(uint64_t size)
    {
        if (size <= TailIdleSize())
            return;
        if (size <= IdleSize())
        {
            // 将有效数据移动到头部
            std::copy(Begin() + _read_pos, Begin() + _write_pos, Begin());
            _write_pos -= _read_pos;
            _read_pos = 0;
            return;
        }
        // 扩容, 不移动数据
        uint64_t new_size = _buffer.size() + size;
        _buffer.resize(new_size);
        assert(_buffer.size() >= new_size);
    }

    void Write(const void *data, uint64_t size)
    {
        if (size == 0)
            return;
        EnsureWriteSize(size);
        assert(TailIdleSize() >= size);
        std::copy(static_cast<const char *>(data), static_cast<const char *>(data) + size, WritePos());
    }

    // 将数据写入缓冲区并移动写指针
    void WriteAndPush(const void *data, uint64_t size)
    {
        Write(data, size);
        MoveWriteOffset(size);
    }

    void WriteString(const std::string &str) { Write(str.data(), str.size()); }

    void WriteStringAndPush(const std::string &str) { WriteAndPush(str.data(), str.size()); }

    void WriteBuffer(const Buffer &buffer) { Write(buffer.ReadPos(), buffer.EffectiveDataSize()); }

    void WriteBufferAndPush(const Buffer &buffer) { WriteAndPush(buffer.ReadPos(), buffer.EffectiveDataSize()); }

    void Read(void *data, uint64_t size)
    {
        if (size == 0)
            return;
        assert(size <= EffectiveDataSize());
        std::copy(ReadPos(), ReadPos() + size, static_cast<char *>(data));
    }

    void ReadAndPop(void *data, uint64_t size)
    {
        Read(data, size);
        MoveReadOffset(size);
    }

    std::string ReadString(uint64_t size)
    {
        std::string str;
        str.resize(size);
        char *p = &str[0];
        Read(p, size);
        return str;
    }

    std::string ReadStringAndPop(uint64_t size)
    {
        std::string str = ReadString(size);
        MoveReadOffset(size);
        return str;
    }

    // 查找换行符
    char *FindCRLF()
    {
        char *p = std::find(Begin() + _read_pos, Begin() + _write_pos, '\n');
        return p;
    }

    std::string GetLineAndPop()
    {
        char *p = FindCRLF();
        if (p == EffectiveEnd())
            return std::string();
        std::string line = ReadStringAndPop(p - ReadPos() + 1); // +1 for '\n'
        return line;
    }

    void Clear()
    {
        _read_pos = 0;
        _write_pos = 0;
    }

private:
    std::vector<char> _buffer;
    uint64_t _read_pos;
    uint64_t _write_pos;
};

class Socket
{
public:
    Socket(int sockfd = -1) : _sockfd(sockfd) {}

    ~Socket() { Close(); }

    int Fd() const { return _sockfd; }

    bool Create()
    {
        if (_sockfd > 0)
        {
            ERROR_LOG("Socket already exists, sockfd: %d", _sockfd);
            return false;
        }
        // IPPROTO_TCP明确表达使用TCP协议, 比默认0更明确
        int sockfd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
        if (sockfd < 0)
        {
            ERROR_LOG("Create socket fail,: %s", strerror(errno));
            return false;
        }
        _sockfd = sockfd;
        return true;
    }

    bool Bind(uint16_t port, const std::string &ip = "0.0.0.0")
    {
        if (CheckExist() == false)
            return false;
        sockaddr_in addr;
        inet_pton(AF_INET, ip.c_str(), &addr.sin_addr);
        addr.sin_family = AF_INET;
        addr.sin_port = htons(port);
        // bind要求传入网络字节序
        int err = bind(_sockfd, reinterpret_cast<sockaddr *>(&addr), sizeof(addr));
        if (err < 0)
        {
            ERROR_LOG("Bind socket fail, sockfd: %d, port: %u, ip: %s, error: %s", _sockfd, port, ip.c_str(), strerror(errno));
            return false;
        }
        INFO_LOG("Bind socket success, sockfd: %d, port: %u, ip: %s", _sockfd, port, ip.c_str());
        return true;
    }

    bool Listen(int backlog = 5)
    {
        if (CheckExist() == false)
            return false;
        int err = listen(_sockfd, backlog);
        if (err < 0)
        {
            ERROR_LOG("Listen socket fail, sockfd: %d, backlog: %d, error: %s", _sockfd, backlog, strerror(errno));
            return false;
        }
        return true;
    }

    bool Connect(uint16_t port, const std::string &ip)
    {
        if (CheckExist() == false)
            return false;
        sockaddr_in addr;
        inet_pton(AF_INET, ip.c_str(), &addr.sin_addr);
        addr.sin_family = AF_INET;
        addr.sin_port = htons(port);
        // bind要求传入网络字节序
        int err = connect(_sockfd, reinterpret_cast<sockaddr *>(&addr), sizeof(addr));
        if (err < 0)
        {
            ERROR_LOG("Connect fail, sockfd: %d, port: %u, ip: %s, error: %s", _sockfd, port, ip.c_str(), strerror(errno));
            return false;
        }
        INFO_LOG("Connect success, sockfd: %d, port: %u, ip: %s", _sockfd, port, ip.c_str());
        return true;
    }

    int Accept()
    {
        if (CheckExist() == false)
            return -1;
        sockaddr_in addr;
        socklen_t addr_len = sizeof(addr);
        int new_sockfd = ::accept(_sockfd, reinterpret_cast<sockaddr *>(&addr), &addr_len);
        if (new_sockfd < 0)
        {
            ERROR_LOG("Accept socket fail, sockfd: %d, error: %s", _sockfd, strerror(errno));
            return -1;
        }
        INFO_LOG("Accept socket success, new sockfd: %d", new_sockfd);
        return new_sockfd;
    }

    ssize_t Recv(void *buf, size_t len, int flags = 0)
    {
        if (CheckExist() == false)
            return -1;
        ssize_t ret = recv(_sockfd, buf, len, flags);
        if (ret <= 0)
        {
            if (errno == EINTR || errno == EAGAIN)
            {
                INFO_LOG("Recv socket interrupted, sockfd: %d, error: %s", _sockfd, strerror(errno));
                return 0; // 中断了，返回0表示没有数据
            }
            ERROR_LOG("Recv socket fail, sockfd: %d, error: %s", _sockfd, strerror(errno));
            return -1;
        }
        return ret;
    }

    ssize_t RecvNoBlock(void *buf, size_t len)
    {
        return Recv(buf, len, MSG_DONTWAIT);
    }

    ssize_t Send(const void *buf, size_t len, int flags = 0)
    {
        if (CheckExist() == false)
            return -1;
        ssize_t ret = send(_sockfd, buf, len, flags);
        if (ret <= 0)
        {
            if (errno == EINTR || errno == EAGAIN)
            {
                INFO_LOG("Send socket interrupted, sockfd: %d, error: %s", _sockfd, strerror(errno));
                return 0; // 中断了，返回0表示没有数据
            }
            if(errno == EPIPE)
            {
                ERROR_LOG("Send socket fail, sockfd: %d, error: %s", _sockfd, strerror(errno));
                abort();
            }
            ERROR_LOG("Send socket fail, sockfd: %d, error: %s", _sockfd, strerror(errno));
            return -1;
        }
        return ret;
    }

    ssize_t SendNoBlock(const void *buf, size_t len)
    {
        return Send(buf, len, MSG_DONTWAIT);
    }

    // 检查socket是否存在
    bool CheckExist()
    {
        if (_sockfd < 0)
        {
            ERROR_LOG("Socket is not open, sockfd: %d", _sockfd);
            return false;
        }
        return true;
    }

    // 关闭socket
    void Close()
    {
        if (_sockfd != -1)
        {
            close(_sockfd);
        }
        _sockfd = -1;
    }

    // 创建服务器socket并绑定到指定端口
    bool CreateServer(uint16_t port, bool is_block = false, const std::string &ip = "0.0.0.0")
    {
        if (Create() == false)
            return false;
        if (is_block == false)
        {
            if (SetNonBlock() == false)
                return false;
        }
        if (ReuseAddress() == false)
            return false;
        if (Bind(port) == false)
            return false;
        if (Listen() == false)
            return false;
        INFO_LOG("Server socket created, sockfd: %d, port: %u", _sockfd, port);
        return true;
    }

    // 创建客户端socket并连接到指定的服务器
    bool CreateClient(uint16_t dst_port, const std::string &dst_ip)
    {
        if (Create() == false)
            return false;
        if (Connect(dst_port, dst_ip) == false)
            return false;
        INFO_LOG("Client socket created, sockfd: %d, dst_port: %u, dst_ip: %s", _sockfd, dst_port, dst_ip.c_str());
        return true;
    }

    // 设置套接字选项---开启地址端口重用
    // 避免服务器重启后端口被占用导致无法绑定
    // 同时开启SO_REUSEPORT选项
    // SO_REUSEPORT允许多个套接字绑定到同一端口

    bool ReuseAddress()
    {
        if (CheckExist() == false)
        {
            ERROR_LOG("Socket is not open, sockfd: %d", _sockfd);
            return false;
        }
        int opt = 1;
        if (setsockopt(_sockfd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)) < 0)
        {
            ERROR_LOG("Set socket option SO_REUSEADDR fail, sockfd: %d, error: %s", _sockfd, strerror(errno));
            return false;
        }
        INFO_LOG("Set socket option SO_REUSEADDR success, sockfd: %d", _sockfd);
        opt = 1;
        if (setsockopt(_sockfd, SOL_SOCKET, SO_REUSEPORT, &opt, sizeof(opt)) < 0)
        {
            ERROR_LOG("Set socket option SO_REUSEPORT fail, sockfd: %d, error: %s", _sockfd, strerror(errno));
            return false;
        }
        INFO_LOG("Set socket option SO_REUSEPORT success, sockfd: %d", _sockfd);
        return true;
    }

    // 设置套接字为非阻塞模式
    bool SetNonBlock()
    {
        if (CheckExist() == false)
        {
            ERROR_LOG("Socket is not open, sockfd: %d", _sockfd);
            return false;
        }
        int fl = fcntl(_sockfd, F_GETFL);
        if (fl < 0)
        {
            ERROR_LOG("Get socket flags fail, sockfd: %d, error: %s", _sockfd, strerror(errno));
            return false;
        }
        fl |= O_NONBLOCK;
        int err = fcntl(_sockfd, F_SETFL, fl);
        if (err < 0)
        {
            ERROR_LOG("Set socket non-block fail, sockfd: %d, error: %s", _sockfd, strerror(errno));
            return false;
        }
        INFO_LOG("Set socket non-block success, sockfd: %d", _sockfd);
        return true;
    }

private:
    int _sockfd;
};

class Epoller;
class EventLoop;
class Channel
{
public:
    using EventCallback = std::function<void(void)>;

    Channel(EventLoop *eventloop, int fd = -1) : _sockfd(fd), _events(0), _revents(0), _eventloop(eventloop)
    {
        if (_sockfd == 1431655825)
        {
            int x;
        }
    }

    int Fd() const { return _sockfd; }

    void SetReadCallback(EventCallback cb) { _read_callback = std::move(cb); }
    void SetWriteCallback(EventCallback cb) { _write_callback = std::move(cb); }
    void SetCloseCallback(EventCallback cb) { _close_callback = std::move(cb); }
    void SetErrorCallback(EventCallback cb) { _error_callback = std::move(cb); }
    void SetAnyCallback(EventCallback cb) { _any_callback = std::move(cb); }

    void SetFd(int fd)
    {
        assert(fd >= 0);
        _sockfd = fd;
    }

    void Setrevents(uint32_t revents)
    {
        _revents = revents;
    }

    uint32_t Getrevents() const { return _revents; };

    uint32_t GetEvents() const { return _events; }

    void EnableRead()
    {
        if (IsWriteEnabled())
            return;
        _events |= EPOLLIN;
        Update();
    }

    void EnableWrite()
    {
        if (IsWriteEnabled())
            return;
        _events |= EPOLLOUT;
        Update();
    }

    bool IsReadEnabled() const { return _events & EPOLLIN; }

    bool IsWriteEnabled() const { return _events & EPOLLOUT; }

    void DisableRead()
    {
        _events &= ~EPOLLIN;
        Update();
    }

    void DisableWrite()
    {
        _events &= ~EPOLLOUT;
        Update();
    }

    // 只是清除对fd的关心事件
    void DisableAll()
    {
        _events = 0;
        Update();
    }

    // 移除监控,直接从epoll移除fd和对fd的事件关心
    void Remove();
    // 把Channel里的fd和events更新到epoll里
    void Update();

    // 只要触发了事件就调用HandleEvent,如何处理事件由具体的回调函数决定
    void HandleEvent()
    {
        // 触发读事件，或者有优先带外数据，或者对端连接半关闭
        if ((_revents & EPOLLIN) || (_revents & EPOLLPRI) || (_revents & EPOLLRDHUP))
        {
            if (_read_callback)
                _read_callback();
            if (_any_callback)
                _any_callback();
        }

        if (_revents & EPOLLOUT)
        {
            if (_any_callback)
                _any_callback();
            if (_write_callback)
                _write_callback();
        }
        else if (_revents & EPOLLERR)
        {
            if (_any_callback)
                _any_callback();

            if (_error_callback)
                _error_callback();
        }
        else if (_revents & EPOLLHUP)
        {
            if (_close_callback)
                _close_callback();
        }
    }

public:
    int _sockfd;
    uint32_t _events;              // 关心的事件
    uint32_t _revents;             // 实际发生的事件
    EventLoop *_eventloop;         // 关联的Epoller对象
    EventCallback _read_callback;  // 读事件回调
    EventCallback _write_callback; // 写事件回调
    EventCallback _close_callback; // 连接断开事件回调
    EventCallback _error_callback; // 错误事件回调
    EventCallback _any_callback;   // 任意事件回调
};

class Epoller
{
public:
    static const int MAX_EVENT_NUM = 1024;
    Epoller(int max_events = MAX_EVENT_NUM)
        : _events(max_events)
    {
        _epoll_fd = epoll_create(1);
        if (_epoll_fd < 0)
        {
            ERROR_LOG("Create epoll fail, error: %s", strerror(errno));
            return;
        }
        INFO_LOG("Epoll created, fd: %d", _epoll_fd);
    }

    // 根据chennel中的fd和关心事件来修改epoll
    void UpdateEvent(Channel *channel)
    {
        if (!IsChannelExists(channel))
        {
            _channels[channel->Fd()] = channel;
            Update(channel, EPOLL_CTL_ADD);
            return;
        }
        int op = EPOLL_CTL_MOD; // 修改事件
        Update(channel, op);
    }

    // 移除channel中的所有事件关心,并且移除channel
    void RemoveEvent(Channel *channel)
    {
        if (!IsChannelExists(channel))
        {
            ERROR_LOG("Channel not exists, fd: %d", channel->Fd());
            return;
        }
        Update(channel, EPOLL_CTL_DEL);
        _channels.erase(channel->Fd());
    }

    // 返回活跃的channel
    void Wait(std::vector<Channel *> &active_channels, int timeout = -1)
    {
        int n = epoll_wait(_epoll_fd, _events.data(), MAX_EVENT_NUM, timeout);
        if (n < 0)
        {
            if (errno == EINTR)
            {
                return;
            }
            ERROR_LOG("epoll_wait fail, errno : %s", strerror(errno));
            return;
        }
        for (int i = 0; i < n; ++i)
        {
            int fd = _events[i].data.fd;
            uint32_t revents = _events[i].events;
            auto it = _channels.find(fd);
            assert(it != _channels.end());
            it->second->Setrevents(revents);
            active_channels.emplace_back(it->second);
        }
    }

private:
    // 对epoll的操作
    void Update(const Channel *channel, int op)
    {
        int fd = channel->Fd();
        epoll_event event;
        event.events = channel->GetEvents();
        event.data.fd = fd;
        int err = epoll_ctl(_epoll_fd, op, fd, &event);
        if (err < 0)
        {
            ERROR_LOG("Epoll ctl fail, fd: %d, op: %d, error: %s", fd, op, strerror(errno));
            return;
        }
    }

    // 判断channel中的事件是否已经被监控
    bool IsChannelExists(const Channel *channel) const
    {
        return _channels.find(channel->Fd()) != _channels.end();
    }

    bool IsWriteReady(const Channel *channel) const
    {
        for (auto &event : _events)
        {
            if (event.data.fd == channel->Fd() && (event.events & EPOLLOUT))
            {
                return true;
            }
        }
        return false;
    }

private:
    int _epoll_fd;
    std::vector<epoll_event> _events;
    std::unordered_map<int, Channel *> _channels; // fd -> Channel
};

#include <cstdint>
#include <functional>
#include <vector>
#include <memory>
#include <unordered_map>
using TaskFunc = std::function<void()>;
using ReleaseFunc = std::function<void()>;
// 类被析构时自动执行任务
class TimerTask
{
private:
    uint64_t _id;         // 任务的唯一编号
    uint32_t _timeout;    // 任务超时时间
    bool _canceled;       // 任务是否被取消
    TaskFunc _task_cb;    // 任务函数
    ReleaseFunc _release; // 用于删除时间轮中_task_map的weak_ptr

public:
    TimerTask(uint64_t id, uint32_t timeout, TaskFunc task_cb, ReleaseFunc release)
        : _id(id)
        , _timeout(timeout)
        , _canceled(false)
        , _task_cb(task_cb)
        , _release(release)
    {
    }

    ~TimerTask()
    {
        if (!_canceled && _task_cb)
        {
            // 如果任务没有被取消且任务函数存在，则执行任务
            _task_cb();
        }
        if (_release)
        {
            _release(); // 执行释放函数
        }
    }

    // 取消任务
    void Cancel()
    {
        _canceled = true; // 标记任务为已取消
    }

    uint32_t DelayTime() const
    {
        return _timeout;
    }
};

class TimeWheel
{
private:
    using TaskPtr = std::shared_ptr<TimerTask>;
    // 为了不影响shared_ptr的引用计数，使用weak_ptr来存储任务
    using WeakTaskPtr = std::weak_ptr<TimerTask>;
    std::unordered_map<uint64_t, WeakTaskPtr> _task_map; // 任务映射表
    std::vector<std::vector<TaskPtr>> _wheel;            // 时间轮的槽
    int _tick;
    int _capacity; // 任务最多可延时多久，以秒为单位
    EventLoop *_event_loop;
    int _timer_fd;
    Channel _timer_fd_channel;

public:
    TimeWheel(EventLoop *event_loop, int capacity = 60)
        : _wheel(capacity) // 初始化时间轮的槽
        ,_tick(0)
        , _capacity(capacity)
        , _event_loop(event_loop)
        , _timer_fd(CreateTimerFd())
        , _timer_fd_channel(event_loop, _timer_fd)
    {
        _timer_fd_channel.SetReadCallback(std::bind(&TimeWheel::OnTime, this));
        _timer_fd_channel.EnableRead(); // 关注读事件
    }

    void AddTask(int task_id, uint32_t timeout, TaskFunc task_cb);

    void TimerRefresh(uint64_t task_id);

    void TimerCancel(uint64_t task_id);

    bool HasTask(uint64_t task_id) const
    {
        return _task_map.find(task_id) != _task_map.end(); // 检查任务是否存在
    }

private:
    void RemoveTask(uint64_t task_id)
    {
        auto it = _task_map.find(task_id);
        if (it != _task_map.end())
        {
            _task_map.erase(it); // 从映射表中删除任务
        }
    }

    int CreateTimerFd()
    {
        int timer_fd = timerfd_create(CLOCK_MONOTONIC, 0);
        if (timer_fd < 0)
        {
            ERROR_LOG("Create timerfd fail, error: %s", strerror(errno));
            abort();
        }
        itimerspec new_value;
        new_value.it_interval.tv_sec = 1;  // 定时器间隔
        new_value.it_interval.tv_nsec = 0; // 纳秒部分
        new_value.it_value.tv_sec = 1;
        new_value.it_value.tv_nsec = 0;
        int err = timerfd_settime(timer_fd, 0, &new_value, nullptr);
        if (err < 0)
        {
            ERROR_LOG("Set timerfd time fail, error: %s", strerror(errno));
            abort(); // 设置失败，直接退出
        }
        return timer_fd;
    }

    uint64_t ReadTimeFd()
    {
        uint64_t times;
        int n = read(_timer_fd, &times, sizeof(uint64_t));
        if (n < 0)
        {
            if (errno == EINTR || errno == EAGAIN)
            {
                INFO_LOG("Read timerfd interrupted or no data, error: %s", strerror(errno));
                return 0; // 中断或无数据
            }
            ERROR_LOG("Read timerfd fail, error: %s", strerror(errno));
            abort(); // 读取失败，直接退出
        }
        return times;
    }

    void OnTime()
    {
        uint64_t times = ReadTimeFd();
        // 可能有事件处理耗时久，已经过了不止一秒
        // timefd的触发也要被epoll监控
        // 只有在线程调用loop.Start()，在循环里时才会处理定时任务
        // 有可能其他文件描述符的梳理时间较长，过了好几秒
        while (times > 0)
        {
            RunTimerTask(); // 每秒钟执行一次时间轮任务
            --times;
        }
    }

    // 执行时间轮中的任务，每秒钟被调用一次
    void RunTimerTask()
    {
        _tick = (_tick + 1) % _capacity; // 移动到下一个槽
        _wheel[_tick].clear();           // 清空当前槽,触发shared_ptr的析构
    }

    // 添加任务到时间轮
    void AddTaskInLoop(int task_id, uint32_t timeout, TaskFunc task_cb)
    {
        if (timeout > _capacity)
        {
            return; // 超出时间轮容量，忽略任务
        }
        std::function<void()> release = std::bind(&TimeWheel::RemoveTask, this, task_id);
        TaskPtr task_ptr = std::make_shared<TimerTask>(task_id, timeout, task_cb, release);
        int pos = (_tick + timeout) % _capacity;    // 计算任务在时间轮中的位置
        _wheel[pos].emplace_back(task_ptr);         // 将任务添加到对应槽中
        _task_map[task_id] = WeakTaskPtr(task_ptr); // 记录任务映射
    }

    // 刷新/延迟定时任务
    void TimerRefreshInLoop(uint64_t task_id)
    {
        auto it = _task_map.find(task_id);
        if (it == _task_map.end())
        {
            return; // 任务不存在
        }
        WeakTaskPtr weak_task = it->second;
        TaskPtr task_ptr = weak_task.lock(); // 获取shared_ptr
        if (task_ptr == nullptr)
        {
            _task_map.erase(it); // 任务已被释放，移除映射
            return;
        }
        int pos = (_tick + task_ptr->DelayTime()) % _capacity; // 计算新的位置
        _wheel[pos].emplace_back(task_ptr);                    // 重新添加到新的槽中
    }

    void TimerCancelInLoop(uint64_t task_id)
    {
        auto it = _task_map.find(task_id);
        if (it == _task_map.end())
        {
            return;
        }
        WeakTaskPtr weak_task = it->second;
        TaskPtr task_ptr = weak_task.lock(); // 获取shared_ptr
        if (task_ptr == nullptr)
        {
            _task_map.erase(it); // 任务已被释放，移除映射
            return;
        }
        task_ptr->Cancel(); // 取消任务
    }
};

class EventLoop
{
public:
    using Task_t = std::function<void()>;

public:
    static int CreateEventFd()
    {
        int event_fd = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK);
        DEBUG_LOG("Create eventfd, fd: %d", event_fd);
        if (event_fd < 0)
        {
            if (errno == EINTR || errno == EAGAIN)
            {
                INFO_LOG("Create eventfd interrupted or no data, error: %s", strerror(errno));
                return -1; // 中断或无数据
            }
            ERROR_LOG("Create eventfd fail, error: %s", strerror(errno));
            abort();
        }
        return event_fd;
    }

    void ReadEventFd()
    {
        uint64_t value;
        ssize_t n = read(_eventfd, &value, sizeof(value));
        DEBUG_LOG("Read eventfd, fd: %d, value: %lu", _eventfd, value);
        if (n < 0)
        {
            if (errno == EINTR || errno == EAGAIN)
            {
                INFO_LOG("Read eventfd interrupted or no data, error: %s", strerror(errno));
                return; // 中断或无数据
            }
            ERROR_LOG("Read eventfd fail, error: %s", strerror(errno));
            abort();
        }
    }

    void WakeUpEventFd()
    {
        uint64_t value = 1;
        ssize_t n = write(_eventfd, &value, sizeof(value));
        DEBUG_LOG("Write eventfd, fd: %d", _eventfd);
        if (n < 0)
        {
            if (errno == EINTR || errno == EAGAIN)
            {
                INFO_LOG("Write eventfd interrupted or no data, error: %s", strerror(errno));
                return; // 中断或无数据
            }
            ERROR_LOG("Write eventfd fail, error: %s", strerror(errno));
            abort();
        }
        INFO_LOG("Wake up eventfd, value: %lu", value);
    }

    void RunAllTasks()
    {
        std::vector<Task_t> tmp;
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _tasks.swap(tmp);
        }
        for (const Task_t &task : tmp)
        {
            if (task)
            {
                task(); // 执行任务
            }
        }
    }

public:
    EventLoop()
        : _thread_id(std::this_thread::get_id())
        , _eventfd(CreateEventFd())
        , _eventfd_channel(std::make_unique<Channel>(this, _eventfd))
        , _time_wheel(this) // 创建时间轮对象并传入当前EventLoop
    {
        _eventfd_channel->SetReadCallback(std::bind(&EventLoop::ReadEventFd, this));
        _eventfd_channel->EnableRead(); // 关注读事件
    }

    // 1.事件监控
    // 2.执行事件回调
    // 3.完成任务
    void Start()
    {
        while (true)
        {
            std::vector<Channel *> active_channels;
            // 监控事件
            _epoller.Wait(active_channels);
            for (Channel *channel : active_channels)
            {
                // 执行事件回调
                channel->HandleEvent();
            }
            // 完成任务
            RunAllTasks();
        }
    }

    bool IsInLoop() const { return std::this_thread::get_id() == _thread_id; }

    // 线程本身执行任务
    void RunInLoop(const Task_t &task)
    {
        if (!IsInLoop())
        {
            PushTask(task);
            return;
        }
        task();
    }

    // 将任务添加到任务队列中
    void PushTask(const Task_t &task)
    {
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _tasks.emplace_back(task);
        }
        // 唤醒无事件发生而导致的阻塞
        WakeUpEventFd();
    }

    void UpdateEvent(Channel *channel)
    {
        _epoller.UpdateEvent(channel);
    }

    void RemoveEvent(Channel *channel)
    {
        _epoller.RemoveEvent(channel);
    }

    void AddTask(int task_id, uint32_t timeout, TaskFunc task_cb)
    {
        _time_wheel.AddTask(task_id, timeout, task_cb);
    }

    void TimerRefresh(uint64_t task_id)
    {
        _time_wheel.TimerRefresh(task_id);
    }

    void TimerCancel(uint64_t task_id)
    {
        _time_wheel.TimerCancel(task_id);
    }

    bool HasTask(uint64_t task_id) const { return _time_wheel.HasTask(task_id); }

private:
    std::thread::id _thread_id; // 记录当前线程ID
    std::vector<Task_t> _tasks; // 任务队列
    std::mutex _mutex;          // 保护任务队列的互斥锁
    int _eventfd;
    std::unique_ptr<Channel> _eventfd_channel; // 事件通道
    Epoller _epoller;
    TimeWheel _time_wheel; // 时间轮对象
};

class LoopThread
{
private:
    std::mutex _mutex;
    std::condition_variable _cond;
    EventLoop *_event_loop; // 事件循环对象
    std::thread _thread;    // 事件循环线程, 对应EventLoop对象的线程

    void ThreadEntry()
    {
        EventLoop loop;
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _event_loop = &loop;
            _cond.notify_all();
        }
        loop.Start();
    }

public:
    LoopThread()
        : _event_loop(nullptr)
        , _thread(std::bind(&LoopThread::ThreadEntry, this))
    {
    }

    EventLoop *GetLoop()
    {
        EventLoop *loop = nullptr;
        {
            std::unique_lock<std::mutex> lock(_mutex);
            // 第二个参数确保的是被唤醒时的状态而不是阻塞条件
            _cond.wait(lock, [&]()
                       { return _event_loop != nullptr; });
            loop = _event_loop; // 获取事件循环对象
        }
        return _event_loop; // 返回事件循环对象
    }
};


class LoopThreadPool
{
public:
    LoopThreadPool(EventLoop* main_loop)
        :_threads_num(0)
        ,_next(0)
        ,_main_loop(main_loop)
    {}


    void SetThreadsNum(int num)
    {
        if (num < 0)
        {
            ERROR_LOG("Thread number must be greater than 0");
            return;
        }
        _threads_num = num;
    }


    void Create_threads()
    {
        if(_threads_num == 0)
            return;
        _threads.resize(_threads_num);
        _loops.resize(_threads_num);
        for (int i = 0; i < _threads_num; ++i)
        {
            _threads[i] = new LoopThread();
            _loops[i] = _threads[i]->GetLoop(); // 获取每个线程的事件循环对象
        }
    }


    // 获取下一个事件循环对象
    EventLoop* NextLoop()
    {
        if(_threads_num == 0)
            return _main_loop;
        _next = (_next + 1) % _threads_num; // 轮询获取下一个线程
        return _loops[_next]; // 返回下一个事件循环对象
    }
private:
    int _threads_num; // 线程池中的线程数量
    int _next;
    EventLoop *_main_loop; // 主线程的事件循环对象
    std::vector<LoopThread*> _threads;
    std::vector<EventLoop*> _loops;
};


// 核心思想:包含父类指针，通过父类指针指向子类对象，子类对象来存储任意类型的数据
class Any
{
private:
    class holder
    {
    public:
        virtual ~holder() {}
        virtual const std::type_info &type() const = 0;

        virtual holder *clone() const = 0;
    };

    template <class T>
    class placeholder : public holder
    {
    public:
        placeholder(const T &val)
            : _val(val)
        {
        }

        virtual ~placeholder() = default;

        virtual const std::type_info &type() const override
        {
            return typeid(T);
        }

        virtual holder *clone() const override
        {
            return new placeholder<T>(_val);
        }

    public:
        T _val;
    };

private:
    holder *_content;

public:
    Any()
        : _content(nullptr)
    {
    }

    ~Any()
    {
        delete _content;
    }

    template <class T>
    Any(const T &val)
    {
        _content = new placeholder<T>(val);
    }

    template <class T>
    T *get()
    {
        assert(typeid(T) == _content->type());
        return &(((placeholder<T> *)_content)->_val);
    }

    const Any &swap(Any &other)
    {
        std::swap(_content, other._content);
        return *this;
    }

    Any(const Any &other)
    {
        if (other._content == nullptr)
        {
            _content = nullptr;
        }
        else
        {
            _content = other._content->clone();
        }
    }

    const Any &operator=(const Any &other)
    {
        if (this == &other)
        {
            return *this;
        }
        Any temp(other);
        swap(temp);
        return *this;
    }

    template <class T>
    const Any &operator=(const T &val)
    {
        Any temp(val);
        swap(temp);
        return *this;
    }
};

class Connection;
typedef enum
{
    DISCONNECTED,
    CONNECTING,
    CONNECTED,
    DISCONNECTING
} ConnStatu;
using PtrConnection = std::shared_ptr<Connection>;
class Connection : public std::enable_shared_from_this<Connection>
{
private:
    uint64_t _id; // 连接的唯一标识符, 同时作为时间轮里的任务id
    int _sockfd;
    bool _enable_inactive_release; // 是否启用非活动连接释放
    Socket _sock;
    ConnStatu _status;      // 连接状态
    Channel _channel;       // 关联的Channel对象
    Buffer _read_buffer;    // 读缓冲区
    Buffer _write_buffer;   // 写缓冲区
    EventLoop *_event_loop; // 关联的EventLoop对象
    Any _context;           // 连接上下文，可以存储任意类型的数据

    using ConnectedCallback = std::function<void(const PtrConnection &)>;
    using MessageCallback = std::function<void(const PtrConnection &, Buffer &)>;
    using ClosedCallback = std::function<void(const PtrConnection &)>;
    using AnyEventCallback = std::function<void(const PtrConnection &)>;
    // 服务器使用者通过服务器模块来设置的四个回调函数
    ConnectedCallback _connected_callback; // 连接建立回调
    MessageCallback _message_callback;     // 业务处理回调
    ClosedCallback _closed_callback;       // 连接关闭回调
    AnyEventCallback _any_event_callback;  // 任意事件回调

    ClosedCallback _ser_closed_callback; // 服务器关闭连接回调
private:
    /*五个channel的事件回调函数*/
    // Channel的读回调，能使用这个回调说明读事件已经就绪，可以直接读
    void HandleReadEvent()
    {
        char buf[65536];
        ssize_t n = _sock.RecvNoBlock(buf, sizeof(buf));
        // RecvNoBlock封装后返回-1包含断开连接等其他出错的情况
        if (n < 0)
        {
            // 读出错了不直接关闭连接，是因为已读取的数据进行处理后仍有可能发送成功
            ShutdownInLoop();
            return;
        }
        _read_buffer.WriteAndPush(buf, n);
        if (_message_callback)
        {
            _message_callback(shared_from_this(), _read_buffer);
        }
    }

    // Channel的写回调，能使用这个回调说明写事件已经就绪，可以直接写
    void HandleWriteEvent()
    {
        ssize_t n = _sock.SendNoBlock(_write_buffer.ReadPos(), _write_buffer.EffectiveDataSize());
        if (n < 0)
        {
            // 关连接前要处理数据，即使不发出也要处理
            // 有些请求可能不需要发送回客户端

            if (_read_buffer.EffectiveDataSize() > 0)
            {
                _message_callback(shared_from_this(), _read_buffer);
            }
            Release();
            return;
        }
        _write_buffer.MoveReadOffset(n);
        if (_write_buffer.EffectiveDataSize() == 0)
        {
            // 暂时没数据发了可关闭写监控
            _channel.DisableWrite();
            if (_status == DISCONNECTING)
            {
                // 完成ShutdownInLoop()未完成的关闭连接任务
                Release();
            }
        }
    }

    void HandleCloseEvent()
    {
        if (_read_buffer.EffectiveDataSize() > 0)
        {
            _message_callback(shared_from_this(), _read_buffer);
        }
        Release();
    }

    void HandleErrorEvent()
    {
        HandleCloseEvent();
    }

    void HandleAnyEvent()
    {
        if (_enable_inactive_release)
            _event_loop->TimerRefresh(_id);
        if (_any_event_callback)
            _any_event_callback(shared_from_this());
    }

    // 并非真正关闭连接，需要判断是否有数据需要发送，是否有数据要处理
    void ShutdownInLoop()
    {
        // 置为待关闭状态
        _status = DISCONNECTING;
        if (_read_buffer.EffectiveDataSize() > 0)
        {
            _message_callback(shared_from_this(), _read_buffer);
        }
        if (_write_buffer.EffectiveDataSize() > 0)
        {
            // 启用写事件监控后，写事件触发调用HandleWrite时
            // 若写缓冲区已经全部发送应该直接调用Release()
            _channel.EnableWrite();
            return;
        }
        // 无数据可发直接关闭连接
        Release();
    }

    // 获取连接后，要进行一些操作
    void EstablishInLoop()
    {
        // 必须是CONNECTING状态
        // 表示底层连接已建立好，但是还未设置好各选项
        assert(_status == CONNECTING);
        _status = CONNECTED; // 设置连接状态为已连接
        _channel.EnableRead();
        if (_connected_callback)
        {
            // 调用连接建立回调函数
            _connected_callback(shared_from_this());
        }
    }

    // 实际关闭连接
    void ReleaseInLoop()
    {
        // 设置连接状态为已断开
        _status = DISCONNECTED;
        // 移除对Channel的所有事件关心
        _channel.Remove();
        // 关闭socket
        _sock.Close();
        _sockfd = -1;

        // 取消定时任务
        if (_enable_inactive_release && _event_loop->HasTask(_id))
            _event_loop->TimerCancel(_id); // 从时间轮中取消定时任务
        // 调用连接关闭回调函数
        if (_closed_callback)
        {
            _closed_callback(shared_from_this());
        }

        // 调用服务器关闭连接回调函数
        if (_ser_closed_callback)
        {
            _ser_closed_callback(shared_from_this());
        }
    }

    // 并非实际发送数据的接口，只是放入缓冲区，然后启动读事件监控
    void SendInLoop(Buffer buf)
    {
        if (_status != CONNECTED)
        {
            return;
        }
        _write_buffer.WriteBufferAndPush(buf);
        _channel.EnableWrite(); // 启用写事件监控
    }

    void EnableInactiveReleaseInLoop(uint64_t seconds)
    {
        if (_enable_inactive_release)
        {
            return; // 已经启用非活动连接释放
        }
        _enable_inactive_release = true;
        if (_event_loop->HasTask(_id))
        {
            _event_loop->TimerRefresh(_id); // 刷新定时任务
            return;
        }
        // 添加定时任务到时间轮中
        _event_loop->AddTask(_id, seconds, std::bind(&Connection::Release, this));
    }

    void CancelInactiveReleaseInLoop()
    {
        if (!_enable_inactive_release)
        {
            return; // 非活动连接释放未启用
        }
        _enable_inactive_release = false;
        if (_event_loop->HasTask(_id))
        {
            _event_loop->TimerCancel(_id); // 从时间轮中取消定时任务
        }
    }

    void UpgradeInLoop(const Any &context, const ConnectedCallback &connected_callback, const MessageCallback &message_callback, const ClosedCallback &closed_callback, const AnyEventCallback &any_event_callback)
    {
        _context = context;                       // 设置连接上下文
        _connected_callback = connected_callback; // 设置连接建立回调
        _message_callback = message_callback;     // 设置业务处理回调
        _closed_callback = closed_callback;       // 设置连接关闭回调
        _any_event_callback = any_event_callback; // 设置任意事件回调
    }

public:
    Connection(EventLoop *loop, uint64_t conn_id, int sockfd)
        : _id(conn_id), _sockfd(sockfd), _enable_inactive_release(false),
          _sock(sockfd), _status(CONNECTING), _channel(loop, sockfd), _event_loop(loop)
    {
        // 设置Channel的事件回调函数
        _channel.SetReadCallback(std::bind(&Connection::HandleReadEvent, this));
        _channel.SetWriteCallback(std::bind(&Connection::HandleWriteEvent, this));
        _channel.SetCloseCallback(std::bind(&Connection::HandleCloseEvent, this));
        _channel.SetErrorCallback(std::bind(&Connection::HandleErrorEvent, this));
        _channel.SetAnyCallback(std::bind(&Connection::HandleAnyEvent, this));
    }

    ~Connection() { DEBUG_LOG("RELEASE CONNECTION:%p", this); }

    void SetConnectedCallback(ConnectedCallback cb) { _connected_callback = std::move(cb); }
    void SetMessageCallback(MessageCallback cb) { _message_callback = std::move(cb); }
    void SetClosedCallback(ClosedCallback cb) { _closed_callback = std::move(cb); }
    void SetAnyEventCallback(AnyEventCallback cb) { _any_event_callback = std::move(cb); }
    void SetServerClosedCallback(ClosedCallback cb) { _ser_closed_callback = std::move(cb); }
    int Fd() { return _sockfd; }
    int Id() { return _id; }
    bool IsConnected() const { return _status == CONNECTED; }
    void SetContext(const Any &context) { _context = context; }
    Any *GetContext() { return &_context; }

    void Establish()
    {
        _event_loop->RunInLoop(std::bind(&Connection::EstablishInLoop, this));
    }

    // 并不实际发送数据
    void Send(const void *data, int len)
    {
        if (data == nullptr || len <= 0 || _status != CONNECTED)
        {
            ERROR_LOG("Invalid data or length, status: %d", _status);
            return;
        }
        Buffer buf;
        buf.WriteAndPush(data, len);
        _event_loop->RunInLoop([&]()
                               { SendInLoop(std::move(buf)); });
    }

    // 并不实际关闭
    void Shutdown()
    {
        _event_loop->RunInLoop(std::bind(&Connection::ShutdownInLoop, this));
    }

    void EnableInactiveRelease(uint64_t seconds)
    {
        _event_loop->RunInLoop(std::bind(&Connection::EnableInactiveReleaseInLoop, this, seconds));
    }

    void CancelInactiveRelease()
    {
        _event_loop->RunInLoop(std::bind(&Connection::CancelInactiveReleaseInLoop, this));
    }

    // 为了避免出现切换协议还未完成就有事件到来
    void Upgrade(const Any &context, const ConnectedCallback &connected_callback, const MessageCallback &message_callback, const ClosedCallback &closed_callback, const AnyEventCallback &any_event_callback)
    {
        assert(_event_loop->IsInLoop());
        _event_loop->RunInLoop(std::bind(&Connection::UpgradeInLoop, this, context, connected_callback, message_callback, closed_callback, any_event_callback));
    }

    //???
    void Release()
    {
        _event_loop->PushTask(std::bind(&Connection::ReleaseInLoop, this));
    }
};

class Accepter
{

private:
    EventLoop *_event_loop;
    Socket _listen_socket;
    Channel _listen_channel;
    using AcceptCallback = std::function<void(int)>;
    AcceptCallback _accept_callback; // 接受新连接的回调函数
private:
    // 处理读事件，接受新连接
    void HandleRead()
    {
        int connfd = _listen_socket.Accept();
        if (connfd < 0)
        {
            ERROR_LOG("Accept new connection fail, error: %s", strerror(errno));
            return;
        }
        INFO_LOG("Accepted new connection, fd: %d", connfd);
        // 调用回调函数处理新连接
        if (_accept_callback)
        {
            _accept_callback(connfd);
        }
    }

    int CreateListenSocket(uint16_t port)
    {
        bool ret = _listen_socket.CreateServer(port);
        assert(ret);
        return _listen_socket.Fd();
    }

public:
    Accepter(EventLoop *loop, uint16_t port)
        : _event_loop(loop), _listen_channel(loop)
    {
        int fd = CreateListenSocket(port);
        _listen_channel.SetFd(fd);
        _listen_channel.SetReadCallback(std::bind(&Accepter::HandleRead, this));
    }

    void SetAcceptCallback(AcceptCallback cb)
    {
        _accept_callback = std::move(cb);
    }

    // 在_accept_callback被设置后才能开始启用读事件监控
    // 避免有新连接到来但是还未设置回调函数
    void Listen()
    {
        _listen_channel.EnableRead(); // 启用读事件监控
    }
};


class TcpServer
{
private:
    uint16_t _port;
    int _threads_num; // 线程池中的线程数量
    //自动增长的id，用于管理连接和定时任务
    uint64_t _next_id;
    int time_out;
    bool _enable_inactive_release;
    EventLoop _baseloop;
    Accepter _accepter; // 接受新连接的Accepter对象
    std::unordered_map<uint64_t, PtrConnection> _connections; // 存储所有连接的映射表  
    LoopThreadPool _thread_pool; // 线程池对象
    
    using ConnectedCallback = std::function<void(const PtrConnection &)>;
    using MessageCallback = std::function<void(const PtrConnection &, Buffer &)>;
    using ClosedCallback = std::function<void(const PtrConnection &)>;
    using AnyEventCallback = std::function<void(const PtrConnection &)>;
    // 服务器使用者通过服务器模块来设置的四个回调函数
    ConnectedCallback _connected_callback; // 连接建立回调
    MessageCallback _message_callback;     // 业务处理回调
    ClosedCallback _closed_callback;       // 连接关闭回调
    ClosedCallback _server_closed_callback;       // 连接关闭回调
    AnyEventCallback _any_event_callback;  // 任意事件回调

private:
    void RunAfterInLoop(int seconds, const TaskFunc& task)
    {
        ++_next_id;
        if (seconds <= 0 || task == nullptr)
        {
            ERROR_LOG("Invalid seconds or task function");
            return;
        }
        // 将任务添加到时间轮中
        _baseloop.AddTask(_next_id, seconds, task);
    }


    void NewConnection(int conn_fd)
    {
        EventLoop* loop = _thread_pool.NextLoop(); // 获取下一个事件循环对象
        ++_next_id; // 自动增长的连接ID
        PtrConnection conn = std::make_shared<Connection>(loop, _next_id, conn_fd);
        conn->SetConnectedCallback(_connected_callback);
        conn->SetMessageCallback(_message_callback);
        conn->SetClosedCallback(_closed_callback);
        conn->SetAnyEventCallback(_any_event_callback);
        conn->SetServerClosedCallback(std::bind(&TcpServer::RemoveConnection, this, std::placeholders::_1));
        if(_enable_inactive_release)
        {
            // 启用非活动连接释放
            conn->EnableInactiveRelease(time_out);
        }
        conn->Establish(); // 建立连接
        _connections[_next_id] = conn; // 将连接存入映射表
    }


    void RemoveConnectionInLoop(const PtrConnection& conn)
    {
        if (conn == nullptr)
        {
            return; // 连接为空，直接返回
        }
        uint64_t conn_id = conn->Id(); // 获取连接ID
        auto it = _connections.find(conn_id);
        if (it != _connections.end())
        {
            _connections.erase(it); // 从映射表中删除连接
        }
    }


    void RemoveConnection(const PtrConnection& conn)
    {
        _baseloop.RunInLoop(std::bind(&TcpServer::RemoveConnectionInLoop, this, conn));
    }


public:
    TcpServer(uint16_t port, int threads_num = 4)
        :_port(port)
        , _threads_num(threads_num)
        , _next_id(0)
        , _enable_inactive_release(false)
        , _baseloop()
        ,_accepter(&_baseloop, port)
        , _thread_pool(&_baseloop)
    {}

    void SetThreadsNum(int num)
    {
        if (num < 0)
        {
            ERROR_LOG("Thread number must be greater than 0");
            return;
        }
        _threads_num = num;
        _thread_pool.SetThreadsNum(num);
    }


    void SetConnectedCallback(ConnectedCallback cb) { _connected_callback = std::move(cb); }
    void SetMessageCallback(MessageCallback cb) { _message_callback = std::move(cb); }
    void SetClosedCallback(ClosedCallback cb) { _closed_callback = std::move(cb); }
    void SetAnyEventCallback(AnyEventCallback cb) { _any_event_callback = std::move(cb); }
    void EnableInactiveRelease(int seconds) { _enable_inactive_release = true; time_out = seconds; }


    void RunAfter(int seconds, const TaskFunc& task)
    {
        _baseloop.RunInLoop(std::bind(&TcpServer::RunAfterInLoop, this, seconds, task));
    }

    void Start()
    {
        _accepter.SetAcceptCallback(std::bind(&TcpServer::NewConnection, this, std::placeholders::_1));
        _accepter.Listen();
        _thread_pool.Create_threads(); // 创建线程池中的线程
        _baseloop.Start(); // 启动主事件循环
    }
};


// 在EventLoop定义后才能实现，将关心事件和fd更新到epoll中
void Channel::Update() { _eventloop->UpdateEvent(this); }
// 从epoll中移除关心事件和fd
void Channel::Remove() { _eventloop->RemoveEvent(this); }

void TimeWheel::AddTask(int task_id, uint32_t timeout, TaskFunc task_cb)
{
    _event_loop->RunInLoop(std::bind(&TimeWheel::AddTaskInLoop, this, task_id, timeout, task_cb));
}

void TimeWheel::TimerCancel(uint64_t task_id)
{
    _event_loop->RunInLoop(std::bind(&TimeWheel::TimerCancelInLoop, this, task_id));
}

void TimeWheel::TimerRefresh(uint64_t task_id)
{
    _event_loop->RunInLoop(std::bind(&TimeWheel::TimerRefreshInLoop, this, task_id));
}


class NetWork
{
public:
    NetWork()
    {
        signal(SIGPIPE, SIG_IGN); // 忽略SIGPIPE信号
    }
};

static NetWork nw;